package cn.pengpeng;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MySQLRunner implements Tool {

    private Configuration conf = new Configuration();

    @Override
    public void setConf(Configuration conf) {
        this.conf = conf;
    }

    @Override
    public Configuration getConf() {
        return this.conf;
    }

    @Override
    public int run(String[] args) throws Exception {

        Job job = Job.getInstance(this.conf, "test-custom-format");

        conf = job.getConfiguration();

        job.setJarByClass(MySQLRunner.class);

        // inputFormat
        conf.set(MySQLInputFormat.MYSQL_INPUT_DRIVER, "com.mysql.jdbc.Driver");
        conf.set(MySQLInputFormat.MYSQL_INPUT_URL, "jdbc:mysql://192.168.100.1:3306/data");
        conf.set(MySQLInputFormat.MYSQL_INPUT_USERNAME, "username");
        conf.set(MySQLInputFormat.MYSQL_INPUT_PASSWORD, "password");
        conf.set(MySQLInputFormat.MYSQL_INPUT_SELECT_COUNT_SQL, "select count(*) from event_logs");
        conf.set(MySQLInputFormat.MYSQL_INPUT_SELECT_RECORD_SQL, "select uid,sid,url,time from event_logs");
        conf.setLong(MySQLInputFormat.MYSQL_INPUT_SPLIT_PRE_SIZE, 10L);
        conf.setClass(MySQLInputFormat.MYSQL_OUTPUT_VALUE_CLASS, MySQLMapperInputValue.class, MySQLInputWritable.class);
        job.setInputFormatClass(MySQLInputFormat.class);

        // map
        job.setMapperClass(MySQLMapper.class);
        job.setMapOutputKeyClass(MySQLMapperOutputKey.class);
        job.setMapOutputValueClass(MySQLMapperOutputValue.class);

        // reduce
        job.setReducerClass(MySQLReducer.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(MySQLReducerOutputValue.class);

        // outputFormat
        conf.set(MySQLOutputFormat.MYSQL_OUTPUT_DRIVER, "com.mysql.jdbc.Driver");
        conf.set(MySQLOutputFormat.MYSQL_OUTPUT_URL, "jdbc:mysql://192.168.100.1:3306/data");
        conf.set(MySQLOutputFormat.MYSQL_OUTPUT_USERNAME, "username");
        conf.set(MySQLOutputFormat.MYSQL_OUTPUT_PASSWORD, "password");
        conf.setInt(MySQLOutputFormat.MYSQL_OUTPUT_BATCH_SIZE, 2);
        job.setOutputFormatClass(MySQLOutputFormat.class);

        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new MySQLRunner(), args);
    }
}
